package com.bottomlord.reactice.streams;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;

import java.util.LinkedList;
import java.util.List;

/**
 * @author ChenYue
 * @date 2021/3/31 22:30
 */
public class SimplePublisher<T> implements Publisher<T> {

    private final List<Subscriber<? super T>> subscribers = new LinkedList<>();

    @Override
    public void subscribe(Subscriber<? super T> s) {
        SubscriptionAdapter<T> subscriptionAdapter = new SubscriptionAdapter<>(s);
        s.onSubscribe(subscriptionAdapter);
        subscribers.add(subscriptionAdapter.getSubscriber());
    }

    public void publish(T data) {
        subscribers.forEach(subscriber -> {
            subscriber.onNext(data);
        });
    }

    /**
     * 1. 初始化广播者{@link SimplePublisher}
     * 2. 广播者设置其订阅者
     *    a. 创建订阅者{@link BusinessSubscriber}，执行构造器时设置订阅上限
     *    b. 设置过程中，创建订阅信息实现{@link SubscriptionAdapter}，并将订阅者通过构造器传入，在构造器中通过装饰器模式增强订阅者
     *    c. 最终实际放入订阅者集合的是装饰后的订阅者{@link DecoratingSubscriber}
     * 3. 广播者迭代订阅者集合，依次执行广播
     *    a. 装饰的订阅者会通过状态来判断订阅后的行为，cancel complete
     *    b. 订阅者做实际的逻辑处理，同时其通过订阅信息{@link SubscriptionAdapter}变更装饰器的状态
     *          i. 订阅者 ->(组合) 订阅信息
     *         ii. 订阅信息 ->(组合) 装饰的订阅者，
     *        iii. 装饰的订阅者 ->(装饰) 订阅者
     *
     * @param args 引导类参数
     */
    public static void main(String[] args) {
        SimplePublisher<Object> publisher = new SimplePublisher<>();
        publisher.subscribe(new BusinessSubscriber<>(10));

        for (int i = 0; i < 10; i++) {
            publisher.publish(i);
        }
    }
}
